GO GRPC自定义注册和解析

package registrar

import (
	"context"
	"fmt"
	"log"
	"time"

	clientv3 "go.etcd.io/etcd/client/v3"
)

type EtcdRegistrar struct {
	client  *clientv3.Client
	leaseId clientv3.LeaseID
}

func NewEtcdRegistrar(endpoints []string) *EtcdRegistrar {
	cli, err := clientv3.New(clientv3.Config{
		Endpoints:   endpoints,
		DialTimeout: 5 * time.Second,
	})

	if err != nil {
		log.Fatalf("Failed to connect to etcd: %v", err)
	}

	return &EtcdRegistrar{
		client: cli,
	}
}
// 往etcd注册服务名-地址
func (r *EtcdRegistrar) Register(serviceName string, addr string, ttl int64) error {
	leaseResp, err := r.client.Grant(context.Background(), ttl)
	if err != nil {
		return err
	}
	r.leaseId = leaseResp.ID

	_, err = r.client.Put(context.Background(), fmt.Sprintf("/%s/%s", serviceName, addr), addr, clientv3.WithLease(r.leaseId))

	return err
}

// Builder creates a resolver that will be used to watch name resolution updates.
type Builder interface {
	// Build creates a new resolver for the given target.
	//
	// gRPC dial calls Build synchronously, and fails if the returned error is
	// not nil.
	Build(target Target, cc ClientConn, opts BuildOptions) (Resolver, error)
	// Scheme returns the scheme supported by this resolver.  Scheme is defined
	// at https://github.com/grpc/grpc/blob/master/doc/naming.md.  The returned
	// string should not contain uppercase characters, as they will not match
	// the parsed target's scheme as defined in RFC 3986.
	Scheme() string
}

// Resolver watches for the updates on the specified target.
// Updates include address updates and service config updates.
type Resolver interface {
	// ResolveNow will be called by gRPC to try to resolve the target name
	// again. It's just a hint, resolver can ignore this if it's not necessary.
	//
	// It could be called multiple times concurrently.
	ResolveNow(ResolveNowOptions)
	// Close closes the resolver.
	Close()
}

package resolver

import (
	"context"
	"fmt"
	"log"
	"strings"
	"time"

	clientv3 "go.etcd.io/etcd/client/v3"
	"google.golang.org/grpc/resolver"
)

type etcdResolver struct {
	serviceName string
	client      *clientv3.Client
}

func NewEtcdResolver(endpoints []string, serviceName string) resolver.Builder {
	cli, err := clientv3.New(clientv3.Config{
		Endpoints:   endpoints,
		DialTimeout: 5 * time.Second,
	})
	if err != nil {
		log.Fatalf("Failed to connect to etcd: %v", err)
	}
	return &etcdResolver{
		serviceName: serviceName,
		client:      cli,
	}
}

func (r *etcdResolver) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
	prefix := fmt.Sprintf("/%s/", r.serviceName)
	log.Println("prefix", prefix)
	resp, err := r.client.Get(context.Background(), prefix, clientv3.WithPrefix())

	if err != nil {
		return nil, err
	}

	var addresses []resolver.Address
	for _, kv := range resp.Kvs {
		addr := strings.TrimPrefix(string(kv.Key), prefix)
		log.Println("addr", addr)
		addresses = append(addresses, resolver.Address{Addr: addr})
	}
    // 调用resolver.ClientConn通知客户端地址改变
	cc.UpdateState(resolver.State{Addresses: addresses})

	return r, nil

}

func (*etcdResolver) Scheme() string                          { return "etcd" }
func (*etcdResolver) ResolveNow(o resolver.ResolveNowOptions) {}
func (*etcdResolver) Close()

// ClientConn 包含解析器用来通知 gRPC ClientConn 任何更新的回调函数。
//
// 这个接口由 gRPC 实现。用户通常不需要全新实现这个接口。像测试这样的情况下，
// 新的实现应该嵌入这个接口。这允许 gRPC 向这个接口添加新的方法。
type ClientConn interface {
    // UpdateState 根据需要更新 ClientConn 的状态。
    //
    // 如果返回错误，解析器应该再次尝试解析目标。解析器应该使用回退计时器
    // 避免对服务器请求过载。如果解析器确信重新解析不会改变结果，
    // 比如因为它是基于监视的解析器，返回的错误可以被忽略。
    //
    // 如果解析出的 State 与上次报告的相同，可以省略调用 UpdateState。
    UpdateState(State) error
    // ReportError 通知 ClientConn 解析器遇到了错误。ClientConn 将通知负载均衡器，
    // 并开始以指数回退的方式调用解析器的 ResolveNow。
    ReportError(error)
    // NewAddress 由解析器调用，用于通知 ClientConn 一组新解析出的地址。
    // 地址列表应该是完整的解析地址列表。
    //
    // 已废弃：请改用 UpdateState。
    NewAddress(addresses []Address)
    // NewServiceConfig 由解析器调用，用于通知 ClientConn 一个新的服务配置。
    // 服务配置应以 json 字符串形式提供。
    //
    // 已废弃：请改用 UpdateState。
    NewServiceConfig(serviceConfig string)
    // ParseServiceConfig 解析提供的服务配置，并返回一个提供解析配置的对象。
    ParseServiceConfig(serviceConfigJSON string) *serviceconfig.ParseResult
}